package com.atguigu.app

import com.alibaba.fastjson.JSON
import com.atguigu.beans.StartUpLog
import com.atguigu.gmall.constances.CommonConstants
import com.atguigu.util.MykafkaUtil.MyKafkaUtil
import com.atguigu.util.RedisUtil
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.phoenix.spark._
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.time.format.DateTimeFormatter
import java.time.{Instant, LocalDate, LocalDateTime, ZoneId}
object DAUApp extends BaseApp {
  override var appName: String = "DAUApp"
  override var internal: Int = 10




  def main(args: Array[String]): Unit = {
    sc=new StreamingContext("local[4]",appName,Seconds(internal))

    sc.sparkContext.setLogLevel("error")
    runApp{
      val ds = MyKafkaUtil.getKafkaStream(CommonConstants.KAFKA_START_LOG, sc)
      //取出kafka中的json,并转换成样例类
      val ds1 = ds.map(record => {
        val startUpLog: StartUpLog = JSON.parseObject(record.value(), classOf[StartUpLog])
        //因为json字符串中缺少天和小时的属性,所以转换成的对象也没有,给每个对象加上
        val formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd")
        val time = LocalDateTime.ofInstant(Instant.ofEpochMilli(startUpLog.ts), ZoneId.of("Asia/Shanghai"))
        val day = time.format(formatter)
        startUpLog.logDate = day
        startUpLog.logHour = time.getHour + ""
        startUpLog

      })
      ds1.count().print
      //同一批次进行去重,因为需要使用rdd算子groupByKey,所以使用transform
      val ds2 = ds1.transform(rdd => {
        //将样例类转换成元组的格式,key使用日期+设备拼接,方便后续按key去重
        val rdd1 = rdd.map(startUpLog => (startUpLog.logDate + "-" + startUpLog.mid, startUpLog))
        //按照key分组,相同key的分到一组,取每个key的value集合的最小值
        val rdd2 = rdd1.groupByKey()
        val rdd3 = rdd2.flatMap {
          case (key, logs) => logs.toList.sortBy(_.ts).take(1)
        }
        rdd3
      })

      ds2.count().print

      //读取redis保存的之前批次的状态对数据进行去重

      val ds3 = removeDuplicateLog(ds2)
      //ds3执行了三次行动算子count,foreachRDD ,考虑将ds3持久化
      ds3.cache()

      ds3.count().print
      //将去重后的数据保存至redis
      ds3.foreachRDD(rdd=>{

        rdd.foreachPartition(partition=>{
          val jedis = RedisUtil.getJedisClient
          partition.foreach(startuplog=>{
            jedis.sadd("DAU:"+startuplog.logDate,startuplog.mid)
          })

          jedis.close()
        })

      })

      //将数据写入hbase
      //想使用phoenix与spark集成的API依赖,需要将phoenix提供的静态方法导入import org.apache.phoenix.spark._
      //不能用new Cofiguration() 加载4个hadoop的配置文件,hbase-site.xml中的内容，不会读取
      //得用HBaseConfiguration.create(),不仅读取4个hadoop的配置文件,还会读取hbase-site.xml中的内容和hbase-default.xml中的内容
      ds3.foreachRDD(rdd=>{
        rdd.saveToPhoenix("GMALL2020_DAU",Seq("MID", "UID", "APPID", "AREA", "OS", "CH", "TYPE", "VS", "LOGDATE", "LOGHOUR", "TS"),
          HBaseConfiguration.create(),Some("hadoop102:2181"))
      })
    }
  }
  /*
          读取redis保存的之前批次的状态对数据进行去重:方法一

            Redis(1,2,3)
                K: DAU:日期
                V: Set

            本批次数据(1,2,4)

            去重后： (4)

            当前批次有N条启动日志， 会从池中借N次连接，还N次，由Redis服务运算N次

            不推荐
       */
  def removeDuplicateLog(ds: DStream[StartUpLog]): DStream[StartUpLog] = {
    ds.filter(startUpLog => {

      val jedis = RedisUtil.getJedisClient

      val boolean = jedis.sismember("DAU:" + startUpLog.logDate, startUpLog.mid)

      jedis.close()
      !boolean
    })
  }
  /*
    读取redis保存的之前批次的状态对数据进行去重:方法二
        以分区为单位连接借连接和归还连接
              mapPartitions:有返回值           后续还会进行计算,选这个
              foreachPartitions:无返回值

              当前批次有N条启动日志，kafka指定的topic有M个分区，
              会从池中借M次连接，还M次，由Redis服务运算N次
   */
  def removeDuplicateLog1(ds: DStream[StartUpLog]): DStream[StartUpLog] = {
    ds.mapPartitions(partition=>{

      val jedis = RedisUtil.getJedisClient

      val filterlogs = partition.filter(startUpLog => {

        !jedis.sismember("DAU:" + startUpLog.logDate, startUpLog.mid)
      })
      jedis.close()

      filterlogs
    })
  }

  /*
  假设redis的并发负载太高了,无法及时处理大量的sismember请求(现实不存在,redis百万ops)
  那么将判断元素是否存在的逻辑,下载到spark端完成

      将下载的set集合使用广播变量广播！
            在SparkStreaming中广播变量分为两类：
            ①从始至终，广播的变量不存在变化的，只需要开始时，广播一次！
            ②广播的变量，随着批次数据而变化，需要每个批次都广播一次！
            如果你写的代码在 DS的算子中，就是每个批次都执行，如果写的代码不再DS的算子中，
            就是在Driver端只执行一次！

      这里使用从始至终只广播一次的写法(是错的,得按批次广播才行,这里只是演示这种写法)

      该写法意为:当前批次有N条启动日志，kafka指定的topic有M个分区，
        会从池中借1次连接，还1次，只向redis发送 1次请求，共发送1次请求，共广播1次变量
   */
  def removeDuplicateLog2(ds: DStream[StartUpLog]): DStream[StartUpLog] = {

    val jedis = RedisUtil.getJedisClient
    //下载集合
    val set = jedis.smembers("DAU:" + LocalDate.now())
    jedis.close()
    //下载的set是只读变量,使用广播变量进行广播
    val br = sc.sparkContext.broadcast(set)

    ds.filter(log=> !br.value.contains(log.mid))

  }

  /*
  假设redis的并发负载太高了,无法及时处理大量的sismember请求(现实不存在,redis百万ops)
  那么将判断元素是否存在的逻辑,下载到spark端完成

      将下载的set集合使用广播变量广播！
            在SparkStreaming中广播变量分为两类：
            ①从始至终，广播的变量不存在变化的，只需要开始时，广播一次！
            ②广播的变量，随着批次数据而变化，需要每个批次都广播一次！
            如果你写的代码在 DS的算子中，就是每个批次都执行，如果写的代码不再DS的算子中，
            就是在Driver端只执行一次！

      这里使用每个批次都广播1次的写法(该写法是正确的,得按批次广播才行)

      每个批次都广播1次,那么广播变量必须在ds算子中,sparkcontext没有实现序列化,不能写在rdd算子中

      每批次都广播变量，广播变量保存在Executor的堆内存，广播次数频繁，是否Executor会OOM？
          不会，Executor会使用LRU的回收策略，自动回收广播变量。

  如果广播变量在executor已经确定使用完毕，可以使用手动释放 unpersist(),如果下次在当前Executor又用到了，
  此时会继续申请广播。

  如果调用destroy() ，如果下次在当前Executor又用到了，无法获取广播。
   */
  def removeDuplicateLog3(ds: DStream[StartUpLog]): DStream[StartUpLog] = {

    ds.transform(rdd=>{
      val jedis = RedisUtil.getJedisClient


      //下载集合
      val set = jedis.smembers("DAU:" + LocalDate.now())
      jedis.close()
      //下载的set是只读变量,使用广播变量进行广播
      val br = sc.sparkContext.broadcast(set)
      rdd.filter(log=> !br.value.contains(log.mid))
    })




  }
}
